package com.yangu.flink.cdc

import com.alibaba.fastjson.JSONObject
import com.ververica.cdc.debezium.DebeziumDeserializationSchema
import io.debezium.data.Envelope
import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
import org.apache.flink.util.Collector
import org.apache.kafka.connect.data.Struct
import org.apache.kafka.connect.source.SourceRecord

class ConsumerDeserialization extends DebeziumDeserializationSchema[String] {

  /**
   * 转换成json字符串
   * {
   *    //库名
   *    "database":"",
   *    //表名
   *    "table":"",
   *    //删除或者修改之前的数据
   *    "before":"json字符串",
   *    //新增或者修改后的数据
   *    "after":"json字符串",
   *    "operation":"操作类型:insert、update、delete、read"
   * }
   *
   * @param record
   * @param out
   */
  override def deserialize(record: SourceRecord, out: Collector[String]): Unit = {
    val result = new JSONObject()
    //这里需要进行转义分隔符.
    val topic = record.topic().split("\\.")
    result.put("database", topic(1))
    result.put("table", topic(2))
    val value = record.value().asInstanceOf[Struct]
    setData("before", value, result)
    setData("after", value, result)

    //获取操作类型 READ DELETE UPDATE CREATE
    var operation = Envelope.operationFor(record).toString.toLowerCase
    //将create修改成insert
    if ("create" == operation) {
      operation = "insert"
    }
    result.put("operation", operation)

    //向收集器发送数据
    out.collect(result.toJSONString)
  }

  private def setData(dataKey: String, value: Struct, result: JSONObject): Unit = {
    val data = value.getStruct(dataKey)
    if (data != null) {
      val fields = data.schema().fields()
      val jsonObject = new JSONObject(fields.size())
      fields.forEach(field => jsonObject.put(field.name(), data.get(field)))
      result.put(dataKey, jsonObject.toJSONString)
    }
  }

  override def getProducedType: TypeInformation[String] = BasicTypeInfo.STRING_TYPE_INFO

}
